package Service

import Config.Constant
import Entity.Model.Anime
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.json4s._
import org.json4s.native.JsonMethods._
import redis.clients.jedis.Jedis

import java.util.Properties
object Project_platform_Proportion {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val proper = new Properties()
    proper.setProperty("bootstrap.servers", Constant.KAFKA_SERVER_BIND)
    proper.setProperty("group.id",Constant.KAFKA_GROUP_ID)
    val data=env
      .addSource(new FlinkKafkaConsumer[String](Constant.KAFKA_TOPIC,new SimpleStringSchema(),proper))
//      .readTextFile("Input/Data.txt")
      .map(data=>{
        implicit val formats=DefaultFormats
        val parseJson=parse(data,useBigDecimalForDouble = true)
        parseJson.extract[Anime]
      })
      .map(_.platform)
      .map((_,1))
      .keyBy(_._1)
      .sum(1)
      .map(data=>{(data._1,data._2.toLong)})
      .addSink(new UserRichSinkFunction)
    env.execute()
  }
  class UserRichSinkFunction extends RichSinkFunction[(String,Long)] {
    var jedis: Jedis = _
    override def open(parameters: Configuration): Unit = {
      jedis = new Jedis(Constant.REDIS_BIND, Constant.REDIS_POST)
      jedis.select(Constant.REDIS_DATABASE)
    }

    override def invoke(in: (String, Long)): Unit = {
      val DataMap = Map(in._1 -> in._2).map { case (k, v) => s"$k=$v" }.mkString("|")
      jedis.publish(Constant.Project_platform_Proportion, DataMap)
      jedis.hincrBy(Constant.Project_platform_Proportion, in._1, in._2)
      println(in)
    }

    override def close(): Unit = {
      jedis.close()
    }
  }
}
